// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: daniel

#include "coin2/base/request/http2/session.h"

#include <cstring>

#include <glog/logging.h>

#include "coin2/base/request/http2/error.h"
#include "coin2/base/request/http2/request_parser.h"

namespace coin2::base::request::http2 {

int GetSendFailToleranceEnv() {
  int fail_send_tolerance = 1;
  char const* fst = getenv("FAIL_SEND_TOLERANCE");
  if (fst) {
    fail_send_tolerance = atoi(fst);
    LOG(INFO) << "Set Fail_Send_Tolerance=" << fail_send_tolerance;
  }
  return fail_send_tolerance;
}

int GetMaxRequestLimitEnv() {
  int max_request_limit = 9000;
  char const* mrl = getenv("MAX_REQUEST_LIMIT");
  if (mrl) {
    max_request_limit = atoi(mrl);
    LOG(INFO) << "Set Max_Request_Limit=" << max_request_limit;
  }
  return max_request_limit;
}

Session::Session(
    boost::asio::io_context* io_ctx,
    boost::asio::ssl::context* tls_ctx,
    const std::string& host,
    int port,
    const std::string& bind_ip)
    : io_ctx_{io_ctx},
      tls_ctx_{tls_ctx},
      host_{host},
      port_{port},
      bind_ip_{bind_ip},
      timer_{*io_ctx},
      req_id_generator_{} {
  // TODO(daniel) - read from config
  send_fail_tolerance_ = GetSendFailToleranceEnv();
  max_requests_limit_ = GetMaxRequestLimitEnv();
  Init();
}

void Session::Init() {
  for (int i = 0; i < connections_.size(); ++i) {
    connections_[i] = AllocConnection();
  }

  timer_.expires_after(std::chrono::seconds(timer_interval_sec_));
  timer_.async_wait(std::bind(&Session::OnTimer, this, std::placeholders::_1));
}

auto Session::AllocConnection() -> std::shared_ptr<Http2Connection> {
  auto conn = std::make_shared<Http2Connection>(io_ctx_, tls_ctx_, host_, port_, bind_ip_);
  conn->SetRequestsLimit(max_requests_limit_);
  conn->SetFailTolerance(send_fail_tolerance_);
  conn->Connect();
  return conn;
}

void Session::OnTimer(const boost::system::error_code& ec) {
  if (ec) {
    return;
  }
  // process suspended connections
  auto iter = suspend_conns_.cbegin();
  auto cend = suspend_conns_.cend();
  while (iter != cend) {
    if ((*iter)->IsCompleted() /* || !(*iter)->IsConnected() */) {
      DLOG(INFO) << "[http2] retire connection [" << (*iter)->ConnectionID() << "]"
                 << (*iter).get();
      (*iter)->Suspend();
      iter = suspend_conns_.erase(iter);
    } else {
      ++iter;
    }
  }
  timer_.expires_after(std::chrono::seconds(timer_interval_sec_));
  timer_.async_wait(std::bind(&Session::OnTimer, this, std::placeholders::_1));
}

auto Session::CurrentConnection(int64_t ts) -> std::shared_ptr<Http2Connection>& {
  {
    current_conn_index_ = (current_conn_index_ + 1) % 2;
    auto& conn = connections_[current_conn_index_];
    if (conn->HitRequestsLimit() || conn->HitFailTolerance() || conn->ConnectionExpired(ts)) {
      LOG(INFO) << "[http2] suspend connection due to limit/error, total sent: "
                << conn->RequestSent() << ". " << conn.get();
      auto tmp = AllocConnection();
      conn.swap(tmp);
      suspend_conns_.emplace(std::move(tmp));
      current_conn_index_ = (current_conn_index_ + 1) % 2;
    } else if (!conn->IsConnected()) {
      current_conn_index_ = (current_conn_index_ + 1) % 2;
    }
  }
  return connections_[current_conn_index_];
}

void Session::AsyncRequest(RequestType req, CallbackType callback, int priority_weight) {
  auto context = std::make_shared<Http2AsioContext>(req_id_generator_.Get());
  context->inner_req = std::move(req);
  context->callback = std::move(callback);
  context->response.raw_request = &context->inner_req;
  context->response.request_ts = GetCurrentTimestamp();
  context->weight = priority_weight;
  AsyncSend(context);
}

void Session::AsyncSend(std::weak_ptr<Http2AsioContext> context) {
  auto ctx = context.lock();
  if (!ctx) {
    LOG(ERROR) << "[http2] failed to send request due to context is expired";
    // normally it's due to timedout,
    // no need to callback here due to it's handled by process timeout
    return;
  }
  DLOG(INFO) << "[http2] new send request [" << ctx->request_id << "], weight=" << ctx->weight;
  req_queue_.push_back(ctx->request_id);
  pending_reqs_.emplace(ctx->request_id, ctx);
  SendImpl();
}

void Session::SendImpl() {
  if (is_sending_) {
    return;
  }
  is_sending_ = true;
  while (!req_queue_.empty()) {
    auto req_id = req_queue_.front();
    req_queue_.pop_front();
    if (pending_reqs_.count(req_id) == 0) {
      LOG(ERROR) << "[http2] failed to send request [" << req_id << "], context is expired";
      continue;
    }
    auto& ctx = pending_reqs_.at(req_id);
    auto& conn = CurrentConnection(ctx->response.request_ts);
    if (!conn || !conn->IsConnected()) {
      LOG_EVERY_N(ERROR, 10) << "[http2] failed to send request [" << req_id
                             << "], no valid connection";
      io_ctx_->post([this, &ctx]() {
        auto ec = boost::system::errc::make_error_code(boost::system::errc::too_many_files_open);
        ProcessRequest(ctx, std::move(ec));
      });
      continue;
    }
    conn->Submit(
        ctx,
        std::bind(&Session::ProcessRequest, this, std::placeholders::_1, std::placeholders::_2));
  }
  is_sending_ = false;
}

void Session::ProcessRequest(
    std::shared_ptr<Http2AsioContext> context,
    const boost::system::error_code& ec) {
  DLOG(INFO) << "[http2] recv response [" << context->request_id << "], weight=" << context->weight;
  if (context->callback && !context->invalidated) {
    context->response.error_code = ec;
    context->response.raw_response = &context->inner_res;
    context->response.response_ts = GetCurrentTimestamp();
    context->callback(context->response.error_code, &context->response);
  }
  if (pending_reqs_.count(context->request_id) > 0) {
    pending_reqs_.erase(context->request_id);
  }
}

int Session::ProcessTimeout(int64_t timeout) {
  if (timeout <= 0) {
    return 0;
  }
  // processs timedout request
  int num_timeout = 0;
  auto cur_ts = GetCurrentTimestamp();
  auto iter = pending_reqs_.cbegin();
  auto cend = pending_reqs_.cend();
  while (iter != cend) {
    if (cur_ts - iter->second->response.request_ts > timeout) {
      DLOG(INFO) << "[http2] request timedout [" << iter->second->request_id << "]";
      DLOG(INFO) << "\nnow=" << cur_ts << "\nreq=" << iter->second->response.request_ts
                 << "\nsub=" << cur_ts - iter->second->response.request_ts
                 << "\ntimeout=" << timeout;
      auto& tmp = iter->second;
      iter = pending_reqs_.erase(iter);
      tmp->Cancel();
      ++num_timeout;
    } else {
      ++iter;
    }
  }

  return num_timeout;
}

}  // namespace coin2::base::request::http2
